package com.atguigu.realtime.app.dwd.db;

import com.atguigu.realtime.app.BaseSQLApp;
import com.atguigu.realtime.common.Constant;
import com.atguigu.realtime.util.SqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lzc
 * @Date 2022/5/21 14:01
 */
public class Dwd_04_DwdTradeCartAddApp extends BaseSQLApp {
    public static void main(String[] args) {
        new Dwd_04_DwdTradeCartAddApp().init(2004, 2, "Dwd_04_DwdTradeCartAddApp");
    }
    
    @Override
    protected void handle(StreamExecutionEnvironment env,
                          StreamTableEnvironment tEnv) {
//        tEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(10));
        // 或
        tEnv.getConfig().getConfiguration().setString("table.exec.state.ttl", "10 second");
        
        // 1. 创建一个动态表与 topic ods_db进行关联
        readOdsDb(tEnv, "Dwd_04_DwdTradeCartAddApp");
        // 2. 读取字典表数据
        readBaseDic(tEnv);
        
        // 3. 过滤出来 加购事实表
        Table cartAdd = tEnv.sqlQuery("select " +
                                        " `data`['id'] id, " +
                                        " `data`['user_id'] user_id, " +
                                        " `data`['sku_id'] sku_id, " +
                                        " if(`type`='insert', " +
                                        "         `data`['sku_num'], " +
                                        "          cast(cast(`data`['sku_num'] as int) - cast(`old`['sku_num'] as int) as string)" +
                                        "    ) sku_num, " +
                                        " `data`['source_id'] source_id," +
                                        " `data`['source_type'] source_type," +
                                        " ts," +
                                        " pt " +
                                        "from ods_db " +
                                        "where `database`='gmall2022' " +
                                        "and `table`='cart_info' " +
                                        "and (" +
                                        "  `type`='insert' " +
                                        "   or (`type`='update'" +
                                        "         and `old`['sku_num'] is not null " +
                                        "         and cast(`data`['sku_num'] as int) > cast(`old`['sku_num'] as int)" +  // 数量增加
                                        "      )" +
                                        ")");
        tEnv.createTemporaryView("cart_add", cartAdd);
        cartAdd.printSchema();
        
        // 4. 加购事实表和字段进行 lookup join
        Table resultTable = tEnv.sqlQuery("select" +
                                        " ca.id, " +
                                        " ca.user_id, " +
                                        " ca.sku_id, " +
                                        " ca.sku_num, " +
                                        " ca.source_id, " +
                                        " bd.dic_name source_type_name, " +
                                        " ts " +
                                        "from cart_add ca " +
                                        "join base_dic FOR SYSTEM_TIME AS OF ca.pt as bd " +
                                        "on ca.source_type=bd.dic_code");
        
        // 5. 把结果写入到dwd层
        tEnv.executeSql("create table dwd_trade_cart_add(" +
                            " id string, " +
                            " user_id string, " +
                            " sku_id string, " +
                            " sku_num string, " +
                            " source_id string, " +
                            " source_type_name string, " +
                            " ts string " +
                            ")" + SqlUtil.getKafkaWriteDDL(Constant.TOPIC_DWD_TRADE_CART_ADD));
        
        resultTable.executeInsert("dwd_trade_cart_add");
        
    }
  
}
